package com.rem.flink.flink10Sql;

import com.rem.flink.flink2Source.Event;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 窗口topN
 * 按窗口分组 并且按照点击量 倒序 选取前N个结果返回
 *
 * @author Rem
 * @date 2022-11-08
 */

public class WindowTopNTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据源，并分配时间戳、生成水位线
        SingleOutputStreamOperator<Event> eventStream = env
                .fromElements(
                        new Event("Alice", "./home", 1000L),
                        new Event("Bob", "./cart", 1000L),
                        new Event("Alice", "./prod?id=1", 25 * 60 * 1000L),
                        new Event("Alice", "./prod?id=4", 55 * 60 * 1000L),
                        new Event("Bob", "./prod?id=5", 3600 * 1000L + 60 * 1000L),
                        new Event("Cary", "./home", 3600 * 1000L + 30 * 60 * 1000L),
                        new Event("Cary", "./prod?id=7", 3600 * 1000L + 59 * 60 * 1000L)
                );

        // 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 将数据流转换成表，并指定时间属性
        Schema build = Schema.newBuilder()
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .column("timestamp", DataTypes.BIGINT())
                .columnByExpression("ts", "CAST(TO_TIMESTAMP(FROM_UNIXTIME(`timestamp`)) AS TIMESTAMP(3))")
                .watermark("ts", "SOURCE_WATERMARK()")
                .build();
        Table eventTable = tableEnv.fromDataStream(eventStream, build);
     //   eventTable.printSchema();

        tableEnv.createTemporaryView("eventTable", eventTable);

        // 子查询 窗口聚合 得到窗口信息和用户 访问次数结果表
        String subQuery = " select window_start, window_end, user, count(url) as cnt " +
                "from table ( " +
                "TUMBLE ( TABLE eventTable , DESCRIPTOR(ts), INTERVAL '1' HOUR )" +
                ") group by window_start, window_end, user";


        //外层查询
        String topNquery = "select window_start, window_end, user,cnt,row_num from " +
                " ( select *,ROW_NUMBER() OVER (" +
                " PARTITION BY window_start, window_end order by cnt desc " +
                " ) as row_num " +
                " from ( " + subQuery + " )" +
                " ) where row_num <=2";

        Table result = tableEnv.sqlQuery(topNquery);
        result.printSchema();
        tableEnv.toDataStream(result).print();


        env.execute();
    }
}
